package net.cassandraview.db;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;

import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.thrift.TException;

/**
 * Connection to database for performing management operations, such as creation/dropping keyspaces/column families etc.
 *
 * User: nagakhl
 * Date: 22.01.12
 * Time: 18:32
 */
public class ThriftManagementConnection extends ThriftConnection implements ManagementConnection {

    @Override
    public String createKeyspace(String keyspaceName) {
        Map<String, String> options = new HashMap<String, String>();
        options.put("replication_factor", "1");
        return createKeyspace(keyspaceName, "org.apache.cassandra.locator.SimpleStrategy", options);
    }

    @Override
    public String createKeyspace(String keyspaceName, String strategyName, Map<String, String> options) {
        final KsDef ksDef = new KsDef();
        ksDef.setFieldValue(KsDef._Fields.NAME, keyspaceName);
        ksDef.setFieldValue(KsDef._Fields.STRATEGY_CLASS, strategyName);
        ksDef.setFieldValue(KsDef._Fields.CF_DEFS, Collections.emptyList());
        ksDef.setFieldValue(KsDef._Fields.STRATEGY_OPTIONS, options);
        return (String) tryExec(new Operation() {
            @Override
            public String execute() throws InvalidRequestException, SchemaDisagreementException, TException {
                return client.system_add_keyspace(ksDef);
            }
        });
    }

    @Override
    public String dropKeyspace(final String keyspaceName) {
        return (String) tryExec(new Operation() {
            @Override
            public String execute() throws InvalidRequestException, SchemaDisagreementException, TException {
                return client.system_drop_keyspace(keyspaceName);
            }
        });
    }

    @Override
    public KsDef getKeyspaceInfo(String keyspaceName) {
        try {
            return client.describe_keyspace(keyspace);
        } catch (Exception e) {
            log.log(Level.SEVERE, "Error retrieving keyspace definition: " + keyspace, e);
            throw new ConnectionException("Error retrieving keyspace definition: " + keyspace, e);
        }
    }

    @Override
    public String createColumnFamily(String name, boolean isSuper) {
        final CfDef cfDef = new CfDef();
        cfDef.setFieldValue(CfDef._Fields.NAME, name);
        cfDef.setFieldValue(CfDef._Fields.KEYSPACE, keyspace);
        cfDef.setFieldValue(CfDef._Fields.COLUMN_TYPE, isSuper ? "Super" : "Standard");
        return (String) tryExec(new Operation() {
            @Override
            public String execute() throws InvalidRequestException, SchemaDisagreementException, TException {
                String result = client.system_add_column_family(cfDef);
                if (result != null) {
                    parentManager.addToSchema(cfDef);
                }
                return result;
            }
        });
    }

    @Override
    public CfDef getColumnFamilyInfo(String name) {
        try {
            KsDef ksDef = client.describe_keyspace(keyspace);
            for (CfDef cfDef : ksDef.getCf_defs()) {
                if (cfDef.getName().equals(name)) {
                    return cfDef;
                }
            }
            return null;
        } catch (Exception e) {
            log.log(Level.SEVERE, "Error retrieving keyspace definition: " + keyspace, e);
            throw new ConnectionException("Error retrieving keyspace definition: " + keyspace, e);
        }
    }

    @Override
    public String dropColumnFamily(final String name) {
        return (String) tryExec(new Operation() {
            @Override
            public String execute() throws InvalidRequestException, SchemaDisagreementException, TException {
                return client.system_drop_column_family(name);
            }
        });
    }

    @Override
    public String createIndex(String columnFamily, String fieldName) {
        return null;
    }

    @Override
    public String dropIndex(String columnFamily, String fieldName) {
        return null;
    }

}
